package com.sunzm.spark.sql

import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.DateFormatUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
 *
 * SparkSQL案例
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-05-31 13:54
 */
object SparkSQLDemo {
  def main(args: Array[String]): Unit = {
    //Logger.getLogger("org").setLevel(Level.WARN)

    val spark: SparkSession = SparkSession
      .builder()
      .appName(s"${this.getClass.getSimpleName.stripSuffix("$")}")
      .master("local[*]")
      .config("spark.default.parallelism", 8)
      .config("spark.sql.shuffle.partitions", 8)
      .getOrCreate()

    //基本查询
    //basicQuery(spark)

    //数字操作示例
    //numberOperation(spark)

    //JSON数据解析以及调用java方法示例
    jsonAndJavaMethod(spark)

    spark.close()
  }

  //基本查询
  def basicQuery(spark: SparkSession) = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    //订单数据
    val orderDF: DataFrame = spark.read.json("data/spark/sql/order.txt")
    //打印表结构
    orderDF.printSchema()
    //注册成视图
    orderDF.createOrReplaceTempView("v_order")
  }

  //数字操作
  def numberOperation(spark: SparkSession) = {

    // ceil/ceiling 向上取整
    //运行结果： 0
    spark.sql("SELECT ceil(-0.1)").show()
    spark.sql("SELECT ceiling(-0.1)").show()

    //运行结果： 4
    spark.sql("SELECT ceiling(3.14)").show()
    spark.sql("SELECT ceiling(3.14)").show()

    //floor 向下取整
    //运行结果： -1
    spark.sql("SELECT floor(-0.1)").show()

    //运行结果： 3
    spark.sql("SELECT floor(3.14)").show()

    //round 四舍五入
    //运行结果： 3
    spark.sql("SELECT round(3.14)").show()
    //运行结果： 4
    spark.sql("SELECT round(3.5)").show()
    //运行结果： 4
    spark.sql("SELECT round(3.8)").show()

    //format_number 数字格式化
    //保留4位小数
    spark.sql("SELECT format_number(12332.123456, 4)")
      .show()
    //将数字格式化为 ##,###.### 的格式
    spark.sql("SELECT format_number(12332.123456, '##,###.###')")
      .show()
  }

  /**
   * 解析JSON数据以及调用Java方法
   * @param spark
   */
  def jsonAndJavaMethod(spark: SparkSession) = {
    //get_json_object 解析JSON的函数
    //订单数据
    val orderDS: Dataset[String] = spark.read.textFile("data/spark/sql/order.txt")

    orderDS.printSchema()

    //注册成视图
    orderDS.createOrReplaceTempView("v_order_string")

    //spark.read.textFile方式读取文件，只要一列，名字叫 value
    //$.prdId[0] 表示 prdId 字段的类型为数组，这里是取第0个元素
    //假设还有个字段是JSONObject类型，例如: "info":{"age":17, "tel": "18810108888"}
    //获取的方式为 $.info.age 、 $.info.tel
    spark.sql(
      """
        |SELECT get_json_object(value, '$.orderId') AS orderId,
        | get_json_object(value, '$.prdId[0]')
        | FROM v_order_string
        |""".stripMargin)
      //.show()

    val orderDF: DataFrame = spark.read.json("data/spark/sql/order.txt")
    orderDF.createOrReplaceTempView("v_order")

    //java_method(class, method[, arg1[, arg2 ..]]) - 使用反射 调用Java的方法
    spark.sql(
      """
        |SELECT orderId, if(orderTime is null, java_method('java.lang.System', 'currentTimeMillis'), orderTime) AS orderTime,
        | java_method('org.apache.commons.lang3.time.DateFormatUtils', 'format',
        | if(orderTime is null, 0, orderTime),
        | 'yyyy-MM-dd HH:mm:ss') AS orderTimeFmt
        | FROM v_order
        |""".stripMargin)
      //.show()

    //对于简单的Java方法调用可以使用java_method，但是对于复杂的java方法，建议使用自定义函数

    //时间戳处理
    val getTimeMillisOrDefault = (orderTime: String) => {
      if(StringUtils.isBlank(orderTime)) System.currentTimeMillis else orderTime.toLong
    }
    //注册自定义函数
    spark.udf.register("getTimeMillisOrDefault", getTimeMillisOrDefault)

    val dateFormat = (orderTime: String) => {
      val fmtTimeMillis = if(StringUtils.isBlank(orderTime)) System.currentTimeMillis else orderTime.toLong
      DateFormatUtils.format(fmtTimeMillis, "yyyy-MM-dd HH:mm:ss")
    }

    spark.udf.register("dateFormatToString", dateFormat)

    spark.sql(
      """
        |SELECT orderId, getTimeMillisOrDefault(orderTime) AS orderTime,
        | dateFormatToString(orderTime) AS orderTimeFmt
        | FROM v_order
        |""".stripMargin)
      .show()
  }
}
